# -*- coding:utf-8 -*-
# @Time   : 2023/5/25 20:28
# @Author : Fushuyue
# @Owner  : YSRD (Insights Value)
import sys
import json
import uuid
import re
import datetime
import pymysql
import redis
import pandas as pd

from flask import Blueprint,jsonify,request,Response


emergencyalert = Blueprint('EmergencyAlert',__name__)
import time


def generate_time_list(start, end):
    start_time = datetime.datetime.strptime(start, "%Y-%m-%d %H:%M:%S")
    end_time = datetime.datetime.strptime(end, "%Y-%m-%d %H:%M:%S")
    print(end_time)
    time_list = []
    while start_time <= end_time:
        timestamp = int(time.mktime(start_time.timetuple()))
        time_list.append(timestamp)
        start_time += datetime.timedelta(minutes=5)
    return time_list

@emergencyalert.route('/EmergencyAlert')
def EmergencyAlert():
    alarm = EmergencyAlarmCenter()
    res=alarm.run()
    return jsonify({"data":res})

# 选择自定义 测试的任务的一次 测试结果
@emergencyalert.route('/EmergencyAlertstream',methods=["GET","POST"])
def EmergencyAlertstream():
    start_time_str = request.args.get('start_time')
    end_time_str = request.args.get('end_time')
    alarm = EmergencyAlarmCenter()
    time_list = generate_time_list(start_time_str, end_time_str)
    data = alarm.run(time_list)
    return jsonify(data)

class EmergencyAlarmCenter():
    def __init__(self):
        # 0. 获取数据库连接
        self.get_db_conn()
        # self.get_test_conn()
        self.get_redis_conn()
        self.local_debug = False  # 用于本地调试代码 true 本地调试，False 线上运行
        self.alarm_table = 'elec_warn_emergency'
        self.ent_cache = dict()  # 缓存各个时段的企业平均用电量
    def calc(self, data_time):
        # 0 查询是否有生效中的预案
        plan_data = self.search_open_plan_info(data_time)
        # 如果有目前所需要生成的预案  则进行 计算 并生成新的报警结果
        if plan_data:
            # 0. 查询在data_time时间已经生成的报警信息
            warns = self.search_warn(data_time, plan_data)
            """
            查询这段时间的报警计划
            """
            warns_prev = self.search_warn_prev(data_time, plan_data)
            # 1 计算减生产线
            # 1.1 获取各生产线的监测数据 在这一个时刻
            line_data = self.get_line_info(plan_data, data_time)
            # 1.3 查看是否有因生产线问题需要报警的企业
            # 生产线哪种情况  才算做 报警呢？？？  记录 需要报警的！！
            need_warn = self.calc_line_warn(line_data, warns, warns_prev)
            # 2. 计算减用电量 和 减用电时长
            # 2.1 查询各预案所对应的上一周期的平均用电量 并保存到对应的预案中
            plan_data = self.search_prev_data(plan_data)
            # 2.2 获取当前时刻的用电量
            other_data = self.get_other_info(data_time, plan_data)
            # 2.3 查看是否有因产量和用电时长问题需要报警的企业
            need_warn = self.calc_other_warn(other_data, need_warn, warns, warns_prev)
            result=self.save_warn(need_warn)
            return result
            # 3. 保存生成的报警信息
        else:
            return "该时刻无报警情况"
    def get_line_info(self, plan_data, data_time):
        search_sql = f"""
        select ecc.plan_id, ecc.control_type, ecc.control_index, ecc.ent_id, ecc.measure, ecc.measure_old,
        ecc.measure_time, eei.ent_name, eei.county_id, eei.county_name, '{data_time}' as data_time,
        edi.id as dev_id, edi.group_other_name, ebi.guid, eei.industry_type_id, eei.town_id,
        ifnull(dit.industry_type, '') as industry_type, ifnull(eei.town_name, '') as town_name
        from elec_control_center ecc
            join elec_dev_info edi on ecc.ent_id=edi.ent_id
            join elecdata_basic_info ebi on edi.id=ebi.devid and ebi.name='总正向有功电度'
            join elec_enterprise_info eei on edi.ent_id=eei.id
            left join dict_industry_type dit on eei.industry_type_id=dit.id
        where ecc.plan_id in ({','.join([str(i['plan_id']) for i in plan_data])})
         and ecc.control_type=1 and ecc.is_deleted = 0 and edi.group_other_name != '';
        """
        self.elec_cur.execute(search_sql)
        base_data = self.elec_cur.fetchall()
        if base_data:
            water_data = self.search_5m_water(data_time, base_data)
            guid_to_value = {i['guid']: i['value'] for i in water_data}
            for i in base_data:
                i['if_open'] = 1 if guid_to_value.get(i['guid'], 0) > 0 else 0

        return base_data if base_data else []
    def search_warn(self, data_time, plan_data):
        # 查询已生成的报警信息
        search_sql = f'''
        select ewe.ent_id, ewe.plan_id from {self.alarm_table} ewe 
        where ewe.is_deleted=0 and warn_time='{data_time}'
        and plan_id in ({','.join([str(i['plan_id']) for i in plan_data])})
        '''
        self.elec_cur.execute(search_sql)
        data = self.elec_cur.fetchall()
        warns = [[int(i['plan_id']), int(i['ent_id'])] for i in data]
        return warns
    # 这里是产生以及生成的报警信息  在这1h时间内  plan_id  以及 ent_id作为企业的匹配对象
    def search_warn_prev(self, data_time, plan_data):
        start_time = (datetime.datetime.strptime(data_time, '%Y-%m-%d %H:%M:%S')
                      - datetime.timedelta(hours=24)).strftime('%Y-%m-%d %H:%M:%S')
        # 查询已生成的报警信息
        search_sql = f'''
        select ewe.ent_id, ewe.plan_id from {self.alarm_table} ewe 
        where ewe.is_deleted=0 and is_first=1 and warn_time<'{data_time}' and warn_time>'{start_time}'
        and plan_id in ({','.join([str(i['plan_id']) for i in plan_data])})
        '''
        self.elec_cur.execute(search_sql)
        data = self.elec_cur.fetchall()
        warns = [[int(i['plan_id']), int(i['ent_id'])] for i in data]
        return warns
    # 查看生产线  需要报警的企业
    def calc_line_warn(self, line_data, warns, warns_prev):
        need_warn = list()
        columns = ['plan_id', 'ent_id', 'control_index', 'measure']
        if line_data:
            # 了解line_data是啥？  获取参与此措施的生产线数据
            # if [int(key[i]) for i in range(0, 2)] in warns: 这部分是检查当前组的前两个标签值（经过转换为整数）构成的列表是否出现在
            # warns
            # 列表中。如果出现，那么意味着这个企业在
            # data_time
            # 时刻已经产生过报警。
            # warns是啥???
            # 如果对应匹配成功 则 代表该企业在这个时间段已经产生过报警
            a=0
            for key, val_df in pd.DataFrame(line_data).groupby(columns):
                a+=1
                if [int(key[i]) for i in range(0, 2)] in warns:
                    continue
                # 了解key[3]到底是什么？  如果没有产生报警  找出 部分停产  放入plan_groups
                matches = key[3].replace('部分停产: ', '').replace('部分停产：', '').split('，')
                # 正则寻找吧数据清晰后放入 plan_groups中
                plan_groups = dict()
                for match in matches:
                    # 如果遇到多种特殊情况
                    if len(match.split('、'))>1:
                        for items in match.split('、'):
                            plan_groups[re.findall(r'[a-zA-Z0-9.]*[\u4e00-\u9fa5]+', items)[0]]=int(re.findall(r'\d+', items)[-1])
                    else:
                        if re.match(r"[a-zA-Z0-9.]*[\u4e00-\u9fa5]+[0-9]+", match):
                            plan_groups[re.findall(r'[a-zA-Z0-9.]*[\u4e00-\u9fa5]+', match)[0]] = int(
                                re.findall(r'\d+', match)[-1])
                        else:
                            plan_groups[match] = -1

                need_warn_flag, content = False, ''
                for group_key, group_df in val_df.groupby(columns + ['group_other_name']):
                    if group_key[4] in plan_groups:
                        need_close_num = plan_groups[group_key[4]]  # 要求停产的设备数
                        # 这里全部停产100%则一定满足于要求 如果该企业没有if_open的长度则说明满足要求
                        # 以及当前停产的设备量<需要停产
                        cur_open_num = len(group_df[group_df['if_open'] == 1])  # 该组内在产的设备数
                            # print(group_df[group_df['if_open'] == 1])
                        if cur_open_num >0:
                            cur_close_num = len(group_df[group_df['if_open'] == 0])  # 该组内停产的设备数
                            if need_close_num > 0 and cur_close_num < need_close_num:
                                content += f'{group_key[4]}要求停产{need_close_num}台，实际停产{cur_close_num}台；'
                                need_warn_flag = True
                            if need_close_num < 0:
                                cur_open_num = len(group_df[group_df['if_open'] == 1])  # 该组内在产的设备数
                                if cur_open_num > 0:
                                    content += f'{group_key[4]}要求停产，实际在产{cur_open_num}台；'
                                    need_warn_flag = True


                if need_warn_flag:
                    warn_info = val_df.to_dict(orient='records')[0]
                    # 该企业在当前时间段(目前默认为1小时)内有报警，则该报警不作为第一次报警，否则作为第一次报警
                    warn_info['is_first'] = 0 if [int(key[i]) for i in range(0, 2)] in warns_prev else 1
                    warn_info['content'] = content
                    need_warn.append(warn_info)
        return need_warn
    # 限产方式2  百分数
    def calc_other_warn(self, other_data, need_warn, warns, warns_prev):
        columns = ['plan_id', 'ent_id']
        for i in other_data:
            prev_df, cur_df = i.get('prev_ent'), i.get('cur_ent')
            if prev_df is not None and cur_df is not None:
                # 当上一周期数据与当前周期数据都存在时
                data_df = pd.merge(cur_df, prev_df, how='inner', on='ent_id')
                for j in data_df.to_dict(orient='records'):
                    if [int(j[k]) for k in columns] in warns:
                        # 该企业在data_time时刻已经产生过该类报警，不再报警
                        continue
                    j['is_first'] = 0 if [int(j[k]) for k in columns] in warns_prev else 1
                    if j['control_type'] == 2 and j['cur_mean_value'] > j['prev_mean_value'] * (
                            j['control_index'] / 100):
                        # 若当前总用电量 > 上周期总用电量 * (指数/100)，该企业需要报警
                        j['content'] = "该企业被要求限产{}%，当前用电量为 {:.2f}kW·h，预案启动前30天平均用电量为 {:.2f}kW·h，" \
                                       "二者比值为 {:.2%}，已超出限定比值".format(j['control_index'],
                                                                     j['cur_mean_value'], j['prev_mean_value'],
                                                                     j['cur_mean_value'] / j['prev_mean_value'])
                        need_warn.append(j)
                    if j['control_type'] == 3:
                        # 企业停产
                        j['content'] = "该企业被要求停产，但其当前用电量为 {:.2f}kW·h".format(j['cur_mean_value'])
                        need_warn.append(j)
        return need_warn

    def process_value(self, value):
        if value is None:
            return 'null'
        if isinstance(value, str):
            value = value.replace("'", "''")
            return f"'{value}'"
        if isinstance(value, int) and value == -99:
            return 'DEFAULT'
        return value
    def search_open_plan_info(self, data_time):
        # 查询生效中的预案
        search_sql = f'''
        select eps.id as start_id, eps.plan_id, eps.start_time, eps.end_time from elec_plan_start eps
        where eps.start_time<='{data_time}' and eps.end_time>'{data_time}' and eps.is_deleted=0 and eps.is_stop=0
        '''
        self.elec_cur.execute(search_sql)
        plan_data = self.elec_cur.fetchall()
        # 遍历所有预案，计算各预案前30天的时间段
        # 如：预案开始时间为2023年5月23日，为期一周，上一周期为2023年5月16日至2023年5月22日
        for i in plan_data:
            st = datetime.datetime.strptime(i['start_time'], '%Y-%m-%d %H:%M:%S')
            i['prev_start'] = (st - datetime.timedelta(days=30)).strftime('%Y-%m-%d %H:%M:%S')
            i['prev_end'] = st.strftime('%Y-%m-%d %H:%M:%S')
        return plan_data if plan_data else []
    def search_prev_data(self, plan_data):
        for i in plan_data:
            key = str(i['plan_id']) + '_' + str(i['start_id'])
            if key in self.ent_cache.keys():
                # 若self.ent_cache中存在该预案启动时段的数据，直接取，无需再查数据库
                i['prev_ent'] = self.ent_cache.get(key)
                continue
            # data_df = self.get_dev_cache(i['plan_id'], i['start_id'], i['prev_start'], i['prev_end'])
            # data_df = self.get_dev_cache(54, 33, '2023-08-30 00:00:00', '2023-09-28 23:59:59')   # 避开使用十月一的数据
            # if data_df is not None:
            #     self.ent_cache[key], i['prev_ent'] = data_df, data_df
            # else:
            #     self.ent_cache[key], i['prev_ent'] = None, None
        return plan_data
    def get_other_info(self, data_time, plan_data):
        for i in plan_data:
            # 1. 查询基本信息   control_type  = 5的情况是未定义的 这里需要注意
            base_sql = f"""
            select ecc.control_type, ecc.control_index, ecc.measure, ecc.measure_old, ecc.measure_time, edi.dev_type,
            eei.id as ent_id, eei.ent_name, eei.industry_type_id, eei.county_id,
            eei.county_name, eei.town_id, eei.town_name, dit.industry_type, ebi.guid
            from elec_control_center ecc
                join elec_dev_info edi on ecc.ent_id=edi.ent_id and edi.dev_type != 4
                join elec_enterprise_info eei on edi.ent_id=eei.id
                join dict_industry_type dit on eei.industry_type_id=dit.id
                join elecdata_basic_info ebi on edi.id=ebi.devid and ebi.name='总正向有功电度'
            where ecc.plan_id={i['plan_id']} and ecc.control_type > 1
            """
            self.elec_cur.execute(base_sql)
            base_data = self.elec_cur.fetchall()
            if base_data:
                water_data = self.search_5m_water(data_time, base_data)
                if water_data:
                    # 3. 计算各企业的用电量
                    cur_data = pd.merge(pd.DataFrame(base_data), pd.DataFrame(water_data), how='left',
                                        on='guid').to_dict(orient='records')
                    columns = ['control_type', 'control_index', 'measure', 'measure_time', 'measure_old', 'ent_id',
                               'ent_name', 'industry_type_id', 'county_id', 'county_name', 'town_id', 'town_name',
                               'industry_type']
                    data_df = self.deal_other_group(cur_data, columns, {'value': 'cur_mean_value'})
                    if data_df is not None:
                        data_df['data_time'] = data_time  # 保存报警时间
                        data_df['plan_id'] = i['plan_id']
                    i['cur_ent'] = data_df
        return plan_data
    def search_5m_water(self, data_time, base_data):
        # 2. 查询5分钟流水数据
        cur_time = datetime.datetime.strptime(data_time, '%Y-%m-%d %H:%M:%S')
        # 11:00:00 - 11:05:00 算作 11:00:00 的用电量, 对应的时间戳需要加上5分钟, 计算将来5分钟的用电量作为当前用电量
        next_time = cur_time + datetime.timedelta(minutes=5)
        cur_table = f'elecdata_data_5m_{cur_time.strftime("%Y%m")}'
        next_table = f'elecdata_data_5m_{next_time.strftime("%Y%m")}'
        water_sql = f"""
        select t1.guid, t2.value-t1.value as value, '{data_time}' as data_time
        from {cur_table} t1 join {next_table} t2 on t1.guid=t2.guid
        where t1.data_time='{data_time}' and t2.data_time='{next_time}'
         and t1.guid in ({','.join([f"'{i['guid']}'" for i in base_data])})
        """
        self.elec_cur.execute(water_sql)
        water_data = self.elec_cur.fetchall()
        water_data = [i for i in water_data if i['value'] > 0.101]    # 剔除低电量的值
        return water_data
    def deal_other_group(self, data, columns, renames):
        if data:
            data_df = pd.DataFrame(data)
            # 1. 有总电的用企业总电计算用电量
            has_total_df = data_df[data_df.dev_type == 1]
            has_total_df = has_total_df.groupby(by=columns + ['data_time']).agg(
                {"value": "sum"}).reset_index()  # 计算每个企业各个时间点的用电量总和
            has_total_df = has_total_df.groupby(by=columns).agg({"value": "mean"}).reset_index().rename(
                columns=renames)  # 计算各个企业每个时间点的平均用电量

            # 2. 无企业总电的用其他设备用电量的和作为用电量
            no_total_df = data_df[(data_df.dev_type > 1) & (~data_df['ent_id'].isin(has_total_df['ent_id'].values))]
            no_total_df = no_total_df.groupby(by=columns + ['data_time']).agg(
                {"value": "sum"}).reset_index()  # 计算每个企业各个时间点的用电量总和
            no_total_df = no_total_df[no_total_df['value'] > 0]  # 剔除用电量为零的时间点
            no_total_df = no_total_df.groupby(by=columns).agg({"value": "mean"}).reset_index().rename(
                columns=renames)  # 计算各个企业每个时间点的平均用电量

            # 3.将有总电和无总电的两类用电量汇总
            data_df = has_total_df.append(no_total_df)
            return data_df
    def save_warn(self, need_warn):
        bak_lst = list()
        for from_data in need_warn:
            to_data = dict()
            if self.local_debug:
                to_data['code'] = str(uuid.uuid4())
            else:
                to_data['code'] = 'EM' + '{:0>14}'.format(self.redis.incr('elec_emerge_warn_code_incr', 1))
            to_data['content'] = from_data.get('content')
            to_data['warn_time'] = str(from_data.get('data_time'))
            to_data['plan_id'] = from_data.get('plan_id')
            to_data['ent_id'] = from_data.get('ent_id')
            to_data['ent_name'] = from_data.get('ent_name')
            to_data['industry_type_id'] = from_data.get('industry_type_id')
            to_data['industry_type'] = from_data.get('industry_type')
            to_data['county_id'] = from_data.get('county_id')
            to_data['county_name'] = from_data.get('county_name')
            to_data['town_id'] = from_data.get('town_id')
            if pd.isna(to_data['town_id']):
                to_data['town_id']="None"
            to_data['town_name'] = from_data.get('town_name')
            to_data['is_first'] = from_data.get('is_first')
            to_data['measure'] = from_data.get('measure')
            to_data['measure_time'] = from_data.get('measure_time')
            to_data['measure_old'] = from_data.get('measure_old')
            bak_lst.append(to_data)
        # 存储到测试库
        # if bak_lst:
        #     # 存储到测试数据库
        #     self.data_to_test_db(self.alarm_table, bak_lst)
        return bak_lst

    # 运行的主程序
    def run(self, time_list=None):
        if time_list is None:
            cur_time = datetime.datetime.now()
            data_time = datetime.datetime(cur_time.year, cur_time.month, cur_time.day, cur_time.hour,
                                           5 * (cur_time.minute // 5), 0) - datetime.timedelta(minutes=20)
            if data_time.hour == 0 and data_time.minute < 20 and data_time.minute >= 0:
                pass   # 每天0点0分到15分不报警
            else:
                data_time = data_time.strftime('%Y-%m-%d %H:%M:00')
            #     获取data_time  这里为一个点 '2024-01-02 17:10:00'
            # data_time='2023-12-28 17:10:00'
            # 进入计算函数
            res=self.calc(data_time)
            return res
        else:
            list_result=[]
            for timestamp in time_list:
                if str(timestamp) == 13:
                    timestamp = timestamp // 1000
                dt = datetime.datetime.fromtimestamp(timestamp)
                # dt = dt - datetime.timedelta(minutes=5)
                res =self.calc(dt.strftime('%Y-%m-%d %H:%M:%S'))
                list_result.append({"time_stamp":dt.strftime('%Y-%m-%d %H:%M:%S'),"result":res})
            return list_result
    def get_dev_cache(self, plan_id, start_id, start_time, end_time):
        # 查询各企业上一周期的用电量
        cache_key = f"elec_plan_prev_average_consumption_{plan_id}_{start_id}"
        print('*' * 20, datetime.datetime.now())
        data_info = self.redis.get(cache_key)
        print('*' * 20, datetime.datetime.now())
        if data_info:
            print(f"使用redis缓存: {cache_key}", sys.getsizeof(data_info))
            return pd.DataFrame(json.loads(data_info, encoding="utf-8"))
        else:
            print(f"没有redis缓存:{cache_key}")
            search_sql = f'''
            select edi.ent_id, eds.value+0 as value, edi.dev_type,
            DATE_FORMAT(eds.data_time, '%Y-%m-%d %H:%i:%S') as data_time
            from elec_control_center ecc
                join elec_enterprise_info eei on ecc.ent_id=eei.id
                join elec_dev_info edi on ecc.ent_id=edi.ent_id and edi.dev_type in (1, 2, 3, 5)
                join elecdata_basic_info ebi on edi.id=ebi.devid and ebi.name='总正向有功电度'
                join elecdata_convert_data as eds on ebi.guid=eds.guid and eds.time_type=1 and eds.value > 0.1
                    and eds.data_time >= '{start_time}' and eds.data_time <= '{end_time}'
            where ecc.control_type > 1 and ecc.is_deleted=0
            '''
            # print(search_sql)
            # print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
            self.elec_cur.execute(search_sql)
            # print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
            prev_data = self.elec_cur.fetchall()
            if prev_data:
                data_df = self.deal_other_group(prev_data, ['ent_id'], {'value': 'prev_mean_value'})
                # 此处用的是15分钟的用电量, 需要再除以3才是5分钟的平均用电量, 因为直接用5分钟数计算, 数据量过大
                data_df['prev_mean_value'] = data_df['prev_mean_value'] / 3
                self.redis.set(cache_key, json.dumps(data_df.to_dict('list')), ex=86400)
                return data_df

        # search_sql = f'''
        #             select edi.ent_id, eds.value+0 as value, edi.dev_type,
        #             DATE_FORMAT(eds.data_time, '%Y-%m-%d %H:%i:%S') as data_time
        #             from elec_control_center ecc
        #                 join elec_enterprise_info eei on ecc.ent_id=eei.id
        #                 join elec_dev_info edi on ecc.ent_id=edi.ent_id and edi.dev_type in (1, 2, 3, 5)
        #                 join elecdata_basic_info ebi on edi.id=ebi.devid and ebi.name='总正向有功电度'
        #                 join elecdata_convert_data as eds on ebi.guid=eds.guid and eds.time_type=1 and eds.value > 0
        #                     and eds.data_time >= '{start_time}' and eds.data_time <= '{end_time}'
        #             where ecc.control_type > 1 and ecc.is_deleted=0
        #             '''
        # print(search_sql)
        # print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
        # self.elec_cur.execute(search_sql)
        # print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
        # prev_data = self.elec_cur.fetchall()
        # if prev_data:
        #     data_df = self.deal_other_group(prev_data, ['ent_id'], {'value': 'prev_mean_value'})
        #     # 此处用的是15分钟的用电量, 需要再除以3才是5分钟的平均用电量, 因为直接用5分钟数计算, 数据量过大
        #     data_df['prev_mean_value'] = data_df['prev_mean_value'] / 3
        #     self.redis.set(cache_key, json.dumps(data_df.to_dict('list')), ex=86400)
        #     print('redis存储完成')
        #     return data_df
    def data_to_test_db(self, table_name, res):
        data_df = pd.DataFrame(res)

        def mysql_replace_into(table, conn, keys, data_iter):
            from sqlalchemy.dialects.mysql import insert
            from sqlalchemy.ext.compiler import compiles
            from sqlalchemy.sql.expression import Insert

            @compiles(Insert)
            def replace_string(insert, compiler, **kw):
                s = compiler.visit_insert(insert, **kw)
                s = s.replace("INSERT INTO", "REPLACE INTO")
                return s

            data = [dict(zip(keys, row)) for row in data_iter]
            conn.execute(table.table.insert(replace_string=""), data)

        print('*' * 30, 'eac向外网数据库的', table_name, '保存数据')
        data_df.to_sql(
            name=table_name,
            con='mysql+pymysql://electricity_api_service:GJlfh7&#jg@mmservice-05.mysql.hotgrid.cn:3306/electricity_data',
            index=False,
            chunksize=500,
            if_exists='append', method=mysql_replace_into)

        if self.local_debug == False:  # 生产环境时需要保存到内网和测试库，本地调试不需要
            print('*' * 30, 'eac向测试数据库的', table_name, '保存数据')
            # 保存到测试数据库
            data_df.to_sql(name=table_name,
                           con='mysql+pymysql://electricity:X1w4G3Hdl7VCNqhs@36.110.47.24:3317/electricity_data_test',
                           index=False,
                           chunksize=500,
                           if_exists='append', method=mysql_replace_into)
            data_df.to_sql(
                name=table_name,
                con='mysql+pymysql://electricity_data:EF2zUl1GHss3yqay@192.168.195.201:3317/electricity_data',
                index=False,
                chunksize=500,
                if_exists='append', method=mysql_replace_into)
    def get_db_conn(self):
        DB_CONN = {
            'host': 'mmservice-05.mysql.hotgrid.cn',
            'user': 'electricity_api_service',
            'password': "GJlfh7&#jg",
            'database': 'electricity_data',
            'port': 3306
        }
        self.elec_conn = pymysql.connect(**DB_CONN)
        self.elec_cur = self.elec_conn.cursor(pymysql.cursors.DictCursor)

    def get_test_conn(self):
        DB_CONN = {
            'host': '36.110.47.24',
            'user': 'electricity',
            'password': "X1w4G3Hdl7VCNqhs",
            'database': 'electricity_data_test',
            'port': 3317
        }
        self.elec_conn = pymysql.connect(**DB_CONN)
        self.elec_cur = self.elec_conn.cursor(pymysql.cursors.DictCursor)

    def get_redis_conn(self):
        self.redis = redis.Redis(
            host='electricity-service.redis.hotgrid.cn',
            port=6379,
            db=2,
            password='phWOgP05ymg01GaR'
        )
    # run2是自定义时间的预案？？？启动
    def run2(self):
        cur_dt = datetime.datetime(2023, 10, 11, 13, 0, 0)
        end_time = datetime.datetime(2023, 10, 11, 15, 0, 0)
        while cur_dt < end_time:
            # data_time = (cur_dt - datetime.timedelta(minutes=5)).strftime('%Y-%m-%d %H:%M:%S')
            # self.calc(data_time)
            self.calc(cur_dt.strftime('%Y-%m-%d %H:%M:%S'))
            cur_dt = cur_dt + datetime.timedelta(minutes=5)
if __name__ == '__main__':
    alarm = EmergencyAlarmCenter()
#     测试这段时间的预案报警
#     alarm.run(time_list='2023-12-28 00:00:00')
    import time
    def generate_time_list(start, end):
        start_time = datetime.datetime.strptime(start, "%Y-%m-%d %H:%M:%S")
        end_time = datetime.datetime.strptime(end, "%Y-%m-%d %H:%M:%S")
        time_list = []
        while start_time <= end_time:
            timestamp = int(time.mktime(start_time.timetuple()))
            time_list.append(timestamp)
            start_time += datetime.timedelta(minutes=5)
        return time_list
#
#     # Generate the time list for specific duration
    start_time_str = '2023-12-27 13:30:00'
    end_time_str = '2023-12-30 11:00:00'
    time_list = generate_time_list(start_time_str, end_time_str)
    # alarm.run()
    alarm.run(time_list)
    alarm.run(time_list)